﻿//
// Copyright (c) 2011 Canyala Innovation AB
//
// Permission is hereby granted, free of charge, to any person obtaining a copy of this software and 
// associated documentation files (the "Software"), to deal in the Software without restriction, including
// without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 
// copies of the Software, and to permit persons to whom the Software is furnished to do so, 
// subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT 
// NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. 
// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, 
// WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH 
// THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. 
//

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;

using Microsoft.WindowsAzure.StorageClient;

using Canyala.Lagoon.Extensions;
using System.Collections.Specialized;
using System.Net;
using System.Globalization;
using Microsoft.WindowsAzure.StorageClient.Protocol;
using Canyala.Lagoon.Reflection;
using Canyala.Lagoon.Contracts;

namespace Canyala.Lagoon.IO
{
    /// <summary>
    /// Implementaion of the Stream interface on top of a windows azure
    /// page blob.
    /// </summary>
    /// <remarks>
    /// The PageBlobStream class is used to enable reading and writing of data from/to
    /// a page blob just like a file. The stream uses an in memory buffer (cache) to 
    /// speed up the data accesses to the blob. It is noteworthy to remember that 
    /// all accesses to the blob store has high latency (around 100ms) but good data
    /// transfer speed, which makes using a relatively large cache optimal.
    /// 
    /// You can find out the max size of the underlying page blob from the stream using the 
    /// MaxLength property. MaxLength is thus the maximum number of bytes you can write to 
    /// the stream.
    /// 
    /// You create a stream in readonly or read/write mode using the static Create() method. 
    /// When opened in read/write mode the PageBlobStream will attempt to acquire a lease on the
    /// underlying blob to lock out all others from writing concurrently to the blob. If this fails (due
    /// to some other process having acquired the lease) an IOException is thrown.
    /// </remarks>
    public sealed class PageBlobStream : Stream
    {
        private long _length;                               // The current length of the stream (stored in the underlying blobs metadata)
        private long _position;                             // The current position of the "cursor"
        private CloudPageBlob _pageBlob;                    // The underlying page blob
        private const string LENGTHMETANAME = "_length";    // The name used for for storing the _length field in blob metadata
        private const int PAGESIZE = 512;                   // The size of pages in the underlying page blob API
        private int _cacheSize;                             // The size of the cache (default 1 MB)
        private byte[] _cache;                              // The cache
        private bool _cacheIsModified;                      // Flag set to true if any bytes have been written to in the current cache and needs to be flushed
        private long _cachePosition;                        // The position into the blob of the first byte of the cache
        private string _leaseId;                            // The lease id aquired to do write operations; null if stream is opened in read only mode
        private bool _isCLosed = false;

        public override bool CanRead
        {
            get { return true; }
        }

        public override bool CanSeek
        {
            get { return true; }
        }

        public override bool CanWrite
        {
            get { return _leaseId != null; }
        }

        /// <summary>
        /// The maximum nymber of bytes that can be written to the stream,
        /// as defined by the underlying blobs size.
        /// </summary>
        public long MaxLength { get { return _pageBlob.Properties.Length; } }

        /// <summary>
        /// Flushes the cache out to the underlying blob.
        /// </summary>
        public override void Flush()
        {
            if (_cacheIsModified)
                WritePages(_cache, _cachePosition, _leaseId);
            _cacheIsModified = false;
        }

        /// <summary>
        /// The current length (in bytes) of the stream.
        /// </summary>
        public override long Length
        {
            get { return _length; }
        }

        /// <summary>
        /// The posiion of the read/write cursor.
        /// </summary>
        public override long Position
        {
            get { return _position; }
            set { _position = value; }
        }

        /// <summary>
        /// Reads one whole cache full of data from the blob into the cache.
        /// </summary>
        /// <param name="offset">The offset into the blob where cache data is read (must be aligned on a 512 byte page boundary).</param>
        /// <remarks>
        /// The cache acts like a window of the data in the underlying blob that is stored
        /// in memory for faster access.
        /// 
        /// CACHE:          |  In memory cache  |
        /// STREAM:     |                                 |
        /// BLOB:       |                                                           |
        /// 
        /// - The cache window is always aligned on a 512 byte blob page boundary and must be
        ///   no larger than 4 MB to be able to write the whole cache out to the blob in one 
        ///   write operation.
        /// - When setting the offset of the cache it must be aligned so that the cache never have any
        ///   part outside the bounds of the underlying blob.
        /// </remarks>
        private void LoadCache(long offset)
        {
            Contract.Requires(offset % PAGESIZE == 0, "Cache offset must be on a page boundary");

            if (_cache == null)
                _cache = new byte[_cacheSize];

            // align the offset so that the cache is inside the blob
            offset = offset > MaxLength - _cacheSize ? MaxLength - _cacheSize : offset;

            Contract.Assume(offset % PAGESIZE == 0, "Cache offset mus be on a page boundary");
            Contract.Assume(offset <= MaxLength - _cacheSize, "Cache must be inbounds of the underlying page blob"); 

            // flush old cache if it contains any changes
            Flush();

            // Load a batch of data from offset into the cache
            using (var reader = _pageBlob.OpenRead())
            {
                reader.Position = offset;
                var bytesRead = 0;
                var bufferOffset = 0;

                while (bufferOffset < _cacheSize && (bytesRead = reader.Read(_cache, bufferOffset, _cacheSize - bufferOffset)) > 0)
                    bufferOffset += bytesRead;

                Contract.Assume(bufferOffset == _cacheSize, "After loading cache bufferOffset should be equal to the cache size");
            }

            // update the cache position
            _cachePosition = offset;

            Contract.Ensures(_cachePosition == offset, "CachePosiion should be set");
        }

        private void WritePages(byte[] pageData, long startPosition, string leaseId)
        {
            Contract.Requires(pageData != null, "pageData != null");
            Contract.Requires(startPosition >= 0 && startPosition % PAGESIZE == 0, "Start position must be zero or greater and aligned on page boundary");

            Dynacutor dyn = new Dynacutor();

            var pageProperties = new PutPageProperties
            {
                PageWrite = PageWrite.Update,
                Range = (PageRange)dyn.CreateInstanceOf(typeof(PageRange), startPosition, (startPosition + pageData.Length - 1))
            };

            string url = _pageBlob.Uri.ToString();
            if (_pageBlob.ServiceClient.Credentials.NeedsTransformUri)
            {
                url = _pageBlob.ServiceClient.Credentials.TransformUri(url);
            }
#if DEBUG
            var req = BlobRequest.PutPage(new Uri(url), 90, pageProperties, null);
#else
            var req = BlobRequest.PutPage(new Uri(url), 90, pageProperties, leaseId);
#endif
            using (var stream = req.GetRequestStream())
            {
                stream.Write(pageData, 0, pageData.Length);
            }
            _pageBlob.ServiceClient.Credentials.SignRequest(req);
            req.GetResponse().Close();    
        }

        public override int Read(byte[] buffer, int offset, int count)
        {
            #region requires
            Contract.Requires(buffer != null, "Buffer must not be null");
            Contract.Requires(buffer.Length >= offset + count, "Data must fit inside the buffer");
            Contract.Requires(offset + count <= buffer.Length, "The sum of offset and count is smaller than the buffer length");
            Contract.Requires(count >= 0, "Count must be positive");
            #endregion

            #region Exceptions
            if (buffer == null)
                throw new ArgumentNullException("buffer is null");
            if (offset < 0 || count < 0)
                throw new ArgumentOutOfRangeException("offset or count is negative.");
            if (offset + count > buffer.Length)
                throw new ArgumentException("The sum of offset and count is larger than the buffer length.");
            if (_isCLosed)
                throw new ObjectDisposedException("stream is closed.");
            #endregion

            var oldPosition = _position;
            count = (int)Math.Min(count, _length - _position);
            var bytesToRead = count > 0 ? count : 0;
            var totalBytesRead = 0;

            if (bytesToRead > 0)
            {
                while (totalBytesRead < bytesToRead)
                {
                    try
                    {
                        var relativeCachePosition = _position - _cachePosition;     // position relative to start of cache

                        if (_cache == null || relativeCachePosition < 0 || relativeCachePosition >= _cacheSize)
                            LoadCache(_position - _position % PAGESIZE);

                        int cacheOffset = (int)(_position - _cachePosition);
                        int cachedBytesToRead = (int)Math.Min(_cacheSize - cacheOffset, bytesToRead);

                        Buffer.BlockCopy(_cache, cacheOffset, buffer, offset, cachedBytesToRead);

                        totalBytesRead += cachedBytesToRead;
                        _position += cachedBytesToRead;
                        offset += cachedBytesToRead;
                    }
                    catch
                    {
                        _position = oldPosition;
                        throw;
                    }
                }
            }

            Contract.Ensures(bytesToRead == _position - oldPosition, "The current position within the stream should be advanced by the number of bytes read");

            return bytesToRead;
        }

        public override long Seek(long offset, SeekOrigin origin)
        {
            switch (origin)
            {
                case SeekOrigin.Begin:
                    _position = offset;
                    break;
                case SeekOrigin.Current:
                    _position = _position + offset;
                    break;
                case SeekOrigin.End:
                    _position = _length - offset;
                    break;
            }

            return _position;
        }

        public override void SetLength(long value)
        {
            Contract.Requires(value <= MaxLength, "Cannot set stream length beyond the blob size");

            _length = value;
            _pageBlob.Metadata[LENGTHMETANAME] = _length.ToString();
        }

        public override void Write(byte[] buffer, int offset, int count)
        {
            #region requires
            Contract.Requires(buffer != null, "Buffer must not be null");
            Contract.Requires(buffer.Length >= offset + count, "Data must fit inside the buffer");
            Contract.Requires(offset + count <= buffer.Length, "The sum of offset and count is smaller than the buffer length");
            Contract.Requires(count >= 0, "Count must be positive");
            #endregion

            if (!CanWrite)
                throw new IOException("You cannot write to a stream that is opened in read only mode");

            if (offset + count > buffer.Length)
                throw new ArgumentOutOfRangeException("buffer", "Buffer length is less than offset + count");

            var bytesToWrite = count;
            var oldPosition = _position;

            try
            {
                while (bytesToWrite > 0)
                {
                    var relativeCachePosition = _position - _cachePosition;     // position relative to start of cache

                    if (_cache == null || relativeCachePosition < 0 || relativeCachePosition >= _cacheSize)
                        LoadCache(_position - _position % PAGESIZE);

                    int cacheOffset = (int)(_position - _cachePosition);
                    int cachedBytesToWrite = (int)Math.Min(_cacheSize - cacheOffset, bytesToWrite);

                    Buffer.BlockCopy(buffer, offset, _cache, cacheOffset, cachedBytesToWrite);

                    _cacheIsModified = true;
                    bytesToWrite -= cachedBytesToWrite;
                    _position += cachedBytesToWrite;
                    offset += cachedBytesToWrite;
                }
            }
            catch
            {
                _position = oldPosition;
                throw;
            }
            finally
            {

            }
            // If we have written data beyond the end of the current length we must update the length
            if (_position > _length)
                SetLength(_position);

            Contract.Ensures(count == _position - oldPosition, "The current position within the stream should be advanced by the number of bytes written");
        }

        private PageBlobStream(CloudPageBlob pageBlob, int cacheSize = 1024 * 1024, bool readOnly = false)
        {
            pageBlob.FetchAttributes();
            _pageBlob = pageBlob;
            _length = 0;

            if (pageBlob.Metadata[LENGTHMETANAME] == null)
                pageBlob.Metadata[LENGTHMETANAME] = _length.ToString();
            else
                _length = pageBlob.Metadata[LENGTHMETANAME].AsLong();

            _position = 0;
            _cacheSize = (int)Math.Min(cacheSize, MaxLength);

            if (!readOnly)
            {   
#if DEBUG
                _leaseId = "123456789";
#else
                var leaseId = _pageBlob.AcquireLease();
                if (!string.IsNullOrEmpty(leaseId))
                    _leaseId = leaseId;
                else
                    throw new IOException("Could not aquire lease on blob: " + pageBlob.ToString());
#endif
            }
        }

        /// <summary>
        /// Creates a new PageBlobStream from a page blob with default settings.
        /// </summary>
        /// <param name="pageBlob">The underlying page blob.</param>
        /// <returns>A new PageBlobStream.</returns>
        public static PageBlobStream Create(CloudPageBlob pageBlob)
        {
            return new PageBlobStream(pageBlob);
        }

        /// <summary>
        /// Creates a new PageBlobStream from a page blob with specified cachce size.
        /// </summary>
        /// <param name="pageBlob">The underlying page blob.</param>
        /// <param name="cacheSize">The size in bytes of the cache (must be multiple of 512 and no larger than 4 MB)</param>
        /// <returns>A new PageBlobStream.</returns>
        public static PageBlobStream Create(CloudPageBlob pageBlob, int cacheSize = 1024 * 1024, bool readOnly = false)
        {
            if (cacheSize % PAGESIZE != 0)
                throw new ArgumentOutOfRangeException("cacheSize", "cache size must be a multiple of azure page blob page size (=512)");
            if (cacheSize > 4 * 1024 * 1024)   // we cant write more than 4 MB in one azure blob write operation!
                throw new ArgumentOutOfRangeException("cacheSize", "cache size must be a smaller than 4 MB");

            return new PageBlobStream(pageBlob, cacheSize, readOnly);
        }

        /// <summary>
        /// Closes the stream and releases the blob lease if in write mode.
        /// </summary>
        public override void Close()
        {
            _isCLosed = true;
            if (CanWrite)
            {
                Flush();
#if DEBUG
                _leaseId = null;          
#else
                _pageBlob.ReleaseLease(_leaseId);
#endif
                _pageBlob.SetMetadata();
            }
        }
    }
}
